package com.sc.nft.helper;


import cn.hutool.core.date.SystemClock;
import cn.hutool.core.lang.Singleton;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.StrUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;

import java.io.Serializable;
import java.util.Date;
import java.util.List;
import java.util.Optional;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * 自增主键生成器，由于目前主要用于分区表生成自增主键使用，应此表和函数都创建于分片库中，注意动态切换数据源时主库是不能使用该类的
 *
 * @author zlikun
 * @since 2024/05/31
 */
@Slf4j
@Component
@RequiredArgsConstructor
public class TableIdGenerator {

//    private final JdbcTemplate jdbcTemplate;
//    private final RedisTemplate<Object, Object> redisTemplate;

    private static final Snowflake snowflake = Singleton.get(Snowflake.class);

    /**
     * 获取下一个主键
     *
     * @param module 功能模块（表名）
     * @param item   条目
     * @return 主键值
     */
    Optional<Long> nextId(String module, String item) {
        return nextIds(module, item, 1).findFirst();
    }

    /**
     * 获取一批主键（有序）
     *
     * @param module 功能模块（表名）
     * @param item   条目
     * @param count  主键数量，取值范围 [1, 1,000,000] | 改用雪花生成器后长度为18位，分片编号（2） + 序列号（16）
     * @return 主键列表（流）
     */
//    @Transactional
    Stream<Long> nextIds(String module, String item, int count) {
        Assert.hasText(module, "模块名称不能为空");
        Assert.hasText(item, "条目名称不能为空");
        Assert.isTrue(count <= 1_000_000, "单次获取主键数量不能超过1,000,000个");

        log.debug("批量获取主键 module: {}, item: {}, count: {}", module, item, count);

//        // 容错处理，未配置Redis客户端时，不走Redis
//        if (redisTemplate != null) {
//            try {
//                return nextIdsFromRedis(module, item, count);
//            } catch (RedisSystemException e) {
//                // 这个分支用于容错处理，后续确认不会调用，则删除之
//                return nextIdsFromRedis2(module, item, count);
//            }
//        } else {
//            return nextIdsFromMySQL(module, item, count);
//        }

        // 改用雪花生成器实现ID序列生成
        return nextIds(count);
    }

    public static Stream<Long> nextIds(int count) {
        return Stream.generate(snowflake::nextId).limit(count);
    }

//    private Stream<Long> nextIdsFromMySQL(String module, String item, int count) {
//
//        // 获取最新主键值
//        Long val = jdbcTemplate.queryForObject("SELECT metago_partition.generate_next_id(?, ?, ?)", Long.class, module, item, count);
//
//        // 条目不存在时返回NULL
//        if (val == null) {
//            log.warn("主键条目({}, {})未初始化", module, item);
//            throw new RuntimeException(String.format("主键条目(%s, %s)未初始化", module, item));
//        }
//
//        // 根据最新主键值和数量生成主键列表
//        return LongStream.range(val - count + 1, val + 1).boxed();
//    }

//    @SneakyThrows
//    private Stream<Long> nextIdsFromRedis(String module, String item, int count) {
//        if (count <= 0) {
//            return Stream.empty();
//        }
//
//        // metago_partition -> mp:pk:$module:$item
//        final String key = this.key(module, item);
//
//        // 按需要获取指定数量主键
//        Stream<Long> ids = this.pop(key, count);
//        if (ids != null) {
//            return ids;
//        }
//
//        // 数量不足时执行初始化，递归重新获取
//        // 这里用幂等控制，保证短时间内初始化逻辑只执行一次（加锁会重复执行）
//        String idempotentKey = String.format("mp:idempotent:%s:%s", module, item);
//        if (BooleanUtil.isTrue(redisTemplate.opsForValue().setIfAbsent(idempotentKey, "1", 5, TimeUnit.SECONDS))) {
//            init(module, item, count);
//        } else {
//            // 未获得初始化机会的请求，等待1秒后再请求
//            TimeUnit.SECONDS.sleep(1L);
//        }
//
//        return nextIdsFromRedis(module, item, count);
//    }

//    /**
//     * 旧实现，LPOP逻辑非原子，改由Lua脚本实现，这里为过渡使用
//     *
//     * @param module
//     * @param item
//     * @param count
//     * @return
//     */
//    @Deprecated
//    @SneakyThrows
//    private Stream<Long> nextIdsFromRedis2(String module, String item, int count) {
//        log.warn("模拟LPOP命令获取ID列表 module: {}, item: {}, count: {}", module, item, count);
//
//        // metago_partition -> mp:pk:$module:$item
//        final String key = this.key(module, item);
//
//        BoundListOperations<Object, Object> operations = redisTemplate.boundListOps(key);
//
//        // 按需要获取指定数量主键
//        List<Object> results;
//        results = operations.range(0, count - 1);  // 批量获取元素
//
//        // 数量不足时执行初始化，递归重新获取
//        if (CollectionUtils.isEmpty(results) || results.size() < count) {
//            // 这里用幂等控制，保证短时间内初始化逻辑只执行一次（加锁会重复执行）
//            String idempotentKey = String.format("mp:idempotent:%s:%s", module, item);
//            if (BooleanUtil.isTrue(redisTemplate.opsForValue().setIfAbsent(idempotentKey, "1", 5, TimeUnit.SECONDS))) {
//                init(module, item, count);
//            } else {
//                // 未获得初始化机会的请求，等待1秒后再请求
//                TimeUnit.SECONDS.sleep(1L);
//            }
//            return nextIdsFromRedis2(module, item, count);
//        } else {
//            // 批量删除元素（LTRIM是保留的元素，因此取剩下的部分） | 模拟批量 pop 动作
//            operations.trim(count, -1);
//        }
//
//        return results.stream().map(id -> {
//            if (id instanceof Number) {
//                return ((Number) id).longValue();
//            } else {
//                return Long.parseLong(id.toString());
//            }
//        });
//    }
//
//    // Redis 6.2+ 支持批量LPOP命令，简化实现
//    static String LUA_6_2 = "local pop_count, total = tonumber(ARGV[1]), redis.call('LLEN', KEYS[1]);" +
//            "return pop_count < total and redis.call('LPOP', KEYS[1], pop_count) or {};";
//
//    // Redis 6.2- 不支持批量LPOP命令，用 LRANGE + LTRIM 模拟实现
//    static final String LUA = "local pop_count, total = tonumber(ARGV[1]), redis.call('LLEN', KEYS[1]); " +
//            "if total < pop_count then return {} end; " +
//            "local items = redis.call('LRANGE', KEYS[1], 0, pop_count-1); " +
//            "redis.call('LTRIM', KEYS[1], pop_count, -1); " +
//            "return items;";
//
//    static final DefaultRedisScript<Object> SCRIPT = new DefaultRedisScript<>(LUA, Object.class);
//
//    static {
//        log.info("Lua script sha1: {}", SCRIPT.getSha1());
//    }
//
//    private Stream<Long> pop(String key, int count) {
//        List<?> items = (List<?>) redisTemplate.execute(SCRIPT, Collections.singletonList(key), String.valueOf(count));
//        if (CollectionUtils.isEmpty(items)) {
//            return null;
//        }
//        return items.stream().map(id -> Long.valueOf(id.toString()));
//    }
//
//    /**
//     * 初始化Redis分区主键池
//     */
//    private void init(String module, String item, int count) {
//        final int BATCH_SIZE = 10000;
//        String key = this.key(module, item);
//        BoundListOperations<Object, Object> operations = redisTemplate.boundListOps(key);
//        // 从数据库中加载 10000 条放入队列中
//        Stream<Long> ids = this.nextIdsFromMySQL(module, item, Math.max(count, BATCH_SIZE));
//        // 追加元素
//        operations.rightPushAll(ids.map(Object::toString).toArray());
//        operations.expire(30, TimeUnit.DAYS);
//
//        log.info("初始化分区主键池 key: {}, size: {}", key, Math.max(count, BATCH_SIZE));
//    }
//
//    private String key(String module, String item) {
//        return String.format("mp:pk:%s:%s", module, item);
//    }

    /**
     * 默认主键生成器
     *
     * @param table     表名
     * @param partition 分区标识
     * @param count     生成数量
     * @return 主键列表
     */
//    @Transactional
    public List<Long> defaultNextIds(String table, String partition, int count) {
        return this.nextIds(table, partition, count)
                .map(id -> String.format("%s%016d", partition, id))
                .map(Long::valueOf)
                .collect(Collectors.toList());
    }

    public List<Long> defaultNextIds(String table, int partitionIndex, int count) {
        return defaultNextIds(table, String.format("%02d", partitionIndex), count);
    }

    /**
     * 从主键提取分片编号
     */
    public static String partition(long id) {

        String str = String.valueOf(id);

        // 旧版ID，第00分片
        if (str.length() < 9) {
            return "00";
        }

        // 旧版ID，单数分片
        if (str.length() == 9) {
            return "0" + str.charAt(0);
        }

        // 新版ID，第00分片
        if (str.length() == 16) {
            return "00";
        }

        // 新版ID，单数分片
        if (str.length() == 17) {
            return "0" + str.charAt(0);
        }

        // 双位分片
        return str.substring(0, 2);
    }

    /**
     * 生成并填充自增主键
     *
     * @param partition 分片（数字）
     * @param items     待填充列表
     * @param generator 主键生成器
     * @param setter    主键设置函数
     * @param <T>       实体类型
     */
    public <T> void fillAutoIncrementIds(String partition, List<T> items,
                                         BiFunction<String, Integer, List<Long>> generator,
                                         BiConsumer<T, Long> setter) {
        if (StrUtil.isBlank(partition) || CollectionUtils.isEmpty(items)) {
            return;
        }
        Assert.notNull(generator, "主键生成器参数不能为空");
        Assert.notNull(setter, "主键设置函数参数不能为空");
        List<Long> ids = generator.apply(partition, items.size());
        for (int i = 0; i < ids.size(); i++) {
            setter.accept(items.get(i), ids.get(i));
        }
    }

    /**
     * 参照 cn.hutool.core.lang.Snowflake 调整实现，标准实现ID长度 19 位，需要留出2位做分片编号用，
     * 故调整代码使生成ID长度16位（Long最大值前两位92不足以表示所有分片，所以再缩短1位）
     *
     * @see cn.hutool.core.lang.Snowflake
     */
    private static class Snowflake implements Serializable {

        private static final long serialVersionUID = 1L;

        /**
         * 默认的起始时间，为Thu, 04 Nov 2010 01:42:54 GMT
         */
        public static long DEFAULT_TWEPOCH = 1288834974657L;
        /**
         * 默认回拨时间，2S
         */
        public static long DEFAULT_TIME_OFFSET = 2000L;

        private static final long WORKER_ID_BITS = 5L;
        // 最大支持机器节点数0~31，一共32个
        @SuppressWarnings({"PointlessBitwiseExpression", "FieldCanBeLocal"})
        private static final long MAX_WORKER_ID = -1L ^ (-1L << WORKER_ID_BITS);
        private static final long DATA_CENTER_ID_BITS = 1L;
        // 最大支持数据中心节点数0~31，一共32个 | DATA_CENTER 当前架构下意义不大，长度缩短为 1 位（不能为0）
        @SuppressWarnings({"PointlessBitwiseExpression", "FieldCanBeLocal"})
        private static final long MAX_DATA_CENTER_ID = -1L ^ (-1L << DATA_CENTER_ID_BITS);
        // 序列号12位（表示只允许workId的范围为：0-4095）| 改为8位（0~255），按当下系统量级足够使用
        private static final long SEQUENCE_BITS = 8L;
        // 机器节点左移12位
        private static final long WORKER_ID_SHIFT = SEQUENCE_BITS;
        // 数据中心节点左移17位
        private static final long DATA_CENTER_ID_SHIFT = SEQUENCE_BITS + WORKER_ID_BITS;
        // 时间毫秒数左移22位
        private static final long TIMESTAMP_LEFT_SHIFT = SEQUENCE_BITS + WORKER_ID_BITS + DATA_CENTER_ID_BITS;
        // 序列掩码，用于限定序列最大值不能超过4095
        @SuppressWarnings("FieldCanBeLocal")
        private static final long SEQUENCE_MASK = ~(-1L << SEQUENCE_BITS);// 4095

        private final long twepoch;
        private final long workerId;
        private final long dataCenterId;
        private final boolean useSystemClock;
        // 允许的时钟回拨数
        private final long timeOffset;

        private long sequence = 0L;
        private long lastTimestamp = -1L;

        /**
         * 构造，使用自动生成的工作节点ID和数据中心ID
         */
        public Snowflake() {
            this(IdUtil.getWorkerId(IdUtil.getDataCenterId(MAX_DATA_CENTER_ID), MAX_WORKER_ID));
        }

        /**
         * 构造
         *
         * @param workerId 终端ID
         */
        public Snowflake(long workerId) {
            this(workerId, IdUtil.getDataCenterId(MAX_DATA_CENTER_ID));
        }

        /**
         * 构造
         *
         * @param workerId     终端ID
         * @param dataCenterId 数据中心ID
         */
        public Snowflake(long workerId, long dataCenterId) {
            this(workerId, dataCenterId, false);
        }

        /**
         * 构造
         *
         * @param workerId         终端ID
         * @param dataCenterId     数据中心ID
         * @param isUseSystemClock 是否使用{@link SystemClock} 获取当前时间戳
         */
        public Snowflake(long workerId, long dataCenterId, boolean isUseSystemClock) {
            this(null, workerId, dataCenterId, isUseSystemClock);
        }

        /**
         * @param epochDate        初始化时间起点（null表示默认起始日期）,后期修改会导致id重复,如果要修改连workerId dataCenterId，慎用
         * @param workerId         工作机器节点id
         * @param dataCenterId     数据中心id
         * @param isUseSystemClock 是否使用{@link SystemClock} 获取当前时间戳
         * @since 5.1.3
         */
        public Snowflake(Date epochDate, long workerId, long dataCenterId, boolean isUseSystemClock) {
            this(epochDate, workerId, dataCenterId, isUseSystemClock, DEFAULT_TIME_OFFSET);
        }

        /**
         * @param epochDate        初始化时间起点（null表示默认起始日期）,后期修改会导致id重复,如果要修改连workerId dataCenterId，慎用
         * @param workerId         工作机器节点id
         * @param dataCenterId     数据中心id
         * @param isUseSystemClock 是否使用{@link SystemClock} 获取当前时间戳
         * @param timeOffset       允许时间回拨的毫秒数
         * @since 5.7.3
         */
        public Snowflake(Date epochDate, long workerId, long dataCenterId, boolean isUseSystemClock, long timeOffset) {
            if (null != epochDate) {
                this.twepoch = epochDate.getTime();
            } else {
                // Thu, 04 Nov 2010 01:42:54 GMT
                this.twepoch = DEFAULT_TWEPOCH;
            }
            if (workerId > MAX_WORKER_ID || workerId < 0) {
                throw new IllegalArgumentException(StrUtil.format("worker Id can't be greater than {} or less than 0", MAX_WORKER_ID));
            }
            if (dataCenterId > MAX_DATA_CENTER_ID || dataCenterId < 0) {
                throw new IllegalArgumentException(StrUtil.format("datacenter Id can't be greater than {} or less than 0", MAX_DATA_CENTER_ID));
            }
            this.workerId = workerId;
            this.dataCenterId = dataCenterId;
            this.useSystemClock = isUseSystemClock;
            this.timeOffset = timeOffset;
        }

        /**
         * 下一个ID
         *
         * @return ID
         */
        public synchronized long nextId() {
            long timestamp = genTime();
            if (timestamp < this.lastTimestamp) {
                if (this.lastTimestamp - timestamp < timeOffset) {
                    // 容忍指定的回拨，避免NTP校时造成的异常
                    timestamp = lastTimestamp;
                } else {
                    // 如果服务器时间有问题(时钟后退) 报错。
                    throw new IllegalStateException(StrUtil.format("Clock moved backwards. Refusing to generate id for {}ms", lastTimestamp - timestamp));
                }
            }

            if (timestamp == this.lastTimestamp) {
                final long sequence = (this.sequence + 1) & SEQUENCE_MASK;
                if (sequence == 0) {
                    timestamp = tilNextMillis(lastTimestamp);
                }
                this.sequence = sequence;
            } else {
                sequence = 0L;
            }

            lastTimestamp = timestamp;

            return ((timestamp - twepoch) << TIMESTAMP_LEFT_SHIFT)
                    | (dataCenterId << DATA_CENTER_ID_SHIFT)
                    | (workerId << WORKER_ID_SHIFT)
                    | sequence;
        }

        // ------------------------------------------------------------------------------------------------------------------------------------ Private method start

        /**
         * 循环等待下一个时间
         *
         * @param lastTimestamp 上次记录的时间
         * @return 下一个时间
         */
        private long tilNextMillis(long lastTimestamp) {
            long timestamp = genTime();
            // 循环直到操作系统时间戳变化
            while (timestamp == lastTimestamp) {
                timestamp = genTime();
            }
            if (timestamp < lastTimestamp) {
                // 如果发现新的时间戳比上次记录的时间戳数值小，说明操作系统时间发生了倒退，报错
                throw new IllegalStateException(
                        StrUtil.format("Clock moved backwards. Refusing to generate id for {}ms", lastTimestamp - timestamp));
            }
            return timestamp;
        }

        /**
         * 生成时间戳
         *
         * @return 时间戳
         */
        private long genTime() {
            return this.useSystemClock ? SystemClock.now() : System.currentTimeMillis();
        }

    }

}
